package com.example.demo.yanglq.component.jedis;

import com.google.common.math.LongMath;
import com.google.common.primitives.Longs;
import org.checkerframework.checker.units.qual.A;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Pipeline;

import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.LongStream;

/**
 * @author by yanglq
 * @Classname RedisBitmaps
 * @Description TODO
 * @Date 2020/1/7/007 14:38
 */
@Component
public class RedisBitmaps {
    private static final String BASE_KEY = "bloomfilter";
    private static final String CURSOR = "cursor";

    private long bitSize;
    @Autowired
    private RedisClient redisClient;
    @Autowired
    private JedisPool jedisPool;

    public void setSize(Long bits) {
        this.bitSize = LongMath.divide(bits, 64, RoundingMode.CEILING) * Long.SIZE;//位数组的长度，相当于n个long的长度
        if (bitCount() == 0) {
            redisClient.bitset(currentKey(), bitSize - 1, false);
        }
    }

    public boolean get(long[] offsets) {
        Jedis jedis = jedisPool.getResource();
        Pipeline pipeline = jedis.pipelined();
        boolean isExisted = true;
        for (long i = 0; i < cursor() + 1; i++) {
            final long cursor = i;
            //只要有一个cursor对应的bitmap中，offsets全部命中，则表示可能存在

            for (long offset : offsets
                    ) {
                pipeline.getbit(genkey(cursor), offset);
            }
        }
        List<Object> results = pipeline.syncAndReturnAll();
        List<Boolean> list = new ArrayList<>();
        results.forEach(obj -> {
            list.add((Boolean) obj);
        });
        for (Boolean result : list) {
            if (!result) {
                isExisted = false;
                break;
            }
        }
        return isExisted;
    }

    boolean get(final long offset) {
        return redisClient.bitget(currentKey(), offset);
    }

    public boolean set(long[] offsets) {
        if (cursor() > 0 && get(offsets)) {
            return false;
        }
        boolean bitsChanged = false;
        for (long offset : offsets) {
            bitsChanged |= set(offset);
        }
        return bitsChanged;
    }

    boolean set(long offset) {
        if (!get(offset)) {
            redisClient.bitset(currentKey(), offset, true);
            return true;
        }
        return false;
    }

    long bitCount() {
        return redisClient.bitCount(currentKey());
    }

    public long bitSize() {
        return this.bitSize;
    }

    private String currentKey() {
        return genkey(cursor());
    }

    private String genkey(long cursor) {
        return BASE_KEY + "-" + cursor;
    }

    private Long cursor() {
        String cursor = null;
        try {
            cursor = redisClient.get(CURSOR);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return cursor == null ? 0 : Longs.tryParse(cursor);
    }

    public void ensureCapacityInternal() {
        if (bitCount() * 2 > bitSize()) {
            grow();
        }
    }

    void grow() {
        Long cursor = redisClient.incr(CURSOR);
        redisClient.bitset(genkey(cursor), bitSize - 1, false);
    }

    public void reset() {
        String[] keys = LongStream.range(0, cursor() + 1).boxed().map(this::genkey).toArray(String[]::new);
        try {
            redisClient.del(keys);
            redisClient.set(CURSOR, "0");
            redisClient.bitset(currentKey(), bitSize - 1, false);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
